package com.king.ml.base

import com.king.ml.enums.TaskNameEnum
import org.apache.spark.SparkConf
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.joda.time.DateTime
import org.springframework.util.StopWatch

import scala.util.{Failure, Success, Try}

trait BaseTask extends Logging with Serializable {

  def taskName: TaskNameEnum.Value
  def initConf(sparkConf: SparkConf = new SparkConf()): SparkConf = sparkConf
  var runtime: StopWatch = _


  def around(implicit spark: SparkSession, currDate: DateTime = DateTime.now): Unit = {
    before
    Try {
      Class.forName(spark.conf.get("task.runner"))
        .newInstance()
        .asInstanceOf[BaseRunner]
        .run
    } match {
      case Success(_) => after
      case Failure(_) => afterThrowException
    }

  }


  private def before(implicit spark: SparkSession, currDate: DateTime): Unit = {

    val taskName = spark.conf.get("task.runner")
    println("开始执行任务 ...["+taskName+"]")
    runtime = new StopWatch(taskName)
    runtime.start(taskName)

  }

  private def after(implicit spark: SparkSession, currDate: DateTime): Unit = {
    val taskName = spark.conf.get("task.runner")
    runtime.stop()
    println("任务执行结束 ...["+ taskName+"],共耗时:" + runtime.getTotalTimeSeconds +"秒")
  }

  private def afterThrowException(implicit spark: SparkSession, currDate: DateTime): Unit = {
    val taskName = spark.conf.get("task.runner")
    runtime.stop()
    println("任务执行异常 ...[" + taskName + "],共耗时:" + runtime.getTotalTimeSeconds + "秒")
  }
}
